package com.ai.fupincore.service.common.impl;

import java.util.HashMap;
import java.util.Map;

import com.ai.frame.bean.InputObject;
import com.ai.frame.util.JsonUtil;
import com.ai.fupincore.service.common.IQueueService;
import com.ai.fupincore.util.Constants.SYSTEM_PROP_CONFIG;
import com.ai.fupincore.util.DedicException;
import com.ai.fupincore.util.PropertiesUtil;
import com.ai.fupincore.util.StringUtil;
import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.client.producer.SendStatus;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.remoting.exception.RemotingException;

public class QueueServiceImpl implements IQueueService {
	private static Map<String, DefaultMQProducer> producerMap = new HashMap<>();
	/**
	 * 获取生产者
	 * @param topic 主题
	 * @return 生产者
	 * @throws MQClientException 
	 */
	private DefaultMQProducer getProducer (String topic) throws DedicException, MQClientException {
		if (producerMap.containsKey(topic)) {
			return producerMap.get(topic);
		}
		DefaultMQProducer producer = new DefaultMQProducer(topic+"_PROD_GROUP");
		producer.setNamesrvAddr(PropertiesUtil.getString(SYSTEM_PROP_CONFIG.REMOTE_MQ_IP_PORT));
		producer.setInstanceName(topic);
		producer.start();
		producerMap.put(topic, producer);
		return producer;
	}

	/**
	 * 将内容放入消息队列
	 * @param topic 主题
	 * @param key 业务标记
	 * @param inputObject 内容
	 * @return 
	 * @throws DedicException
	 * @throws MQClientException 
	 * @throws InterruptedException 
	 * @throws MQBrokerException 
	 * @throws RemotingException 
	 */
	public boolean sendMsg2MQ(String topic, String key, InputObject inputObject) throws DedicException, MQClientException, RemotingException, MQBrokerException, InterruptedException {
		if (StringUtil.isEmpty(topic) || inputObject == null) {
			throw new DedicException("放入队列时,入参为空!");
		}
		DefaultMQProducer producer = getProducer(topic);
		Message msg = new Message(topic, JsonUtil.inputObject2Json(inputObject).getBytes());
		SendResult sendResult = producer.send(msg);
		return SendStatus.SEND_OK.equals(sendResult.getSendStatus());
	}
}
